﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellBlockingAcceptorThread.h"
#include "CellBlockingAcceptor.h"
#include "CellSession.h"
#include "CellMessage.h"
#include "CellLoggerManager.h"

namespace cell
{

#define PARSE_PHASE_UNKNOWN 0
#define PARSE_PHASE_DISTINGUISH 1
#define PARSE_PHASE_EXTRACT 2

#define BA_DEFAULT_TIMEOUT 30000
#define BA_CACHESIZE 8192
#define BA_BUFSIZE 10240

//-----------------------------------------------------------------------
BADaemonThread::BADaemonThread(BlockingAcceptor* acceptor)
	: mAcceptor(acceptor)
	, mRunning(false)
	, mSpinning(false)
{
}
//-----------------------------------------------------------------------
BADaemonThread::~BADaemonThread(void)
{
}
//-----------------------------------------------------------------------
void BADaemonThread::run(void)
{
	mRunning = true;

	mSpinning = true;
	while (mSpinning)
	{
		sockaddr_storage clientaddr;
		int addrlen = sizeof(clientaddr);
		socket_t recver;

		if (INVALID_SOCKET == (recver = ::accept(mAcceptor->mListener, (sockaddr*)&clientaddr, &addrlen)))
		{
			continue;
		}

		// 检查门闩
		if (mAcceptor->latch)
		{
			closeSocket(recver);
			continue;
		}

		// 检查是否达到最大连接数
		if (mAcceptor->getConnectNum() >= mAcceptor->mMaxConnectNum)
		{
			char log[128] = {0x0};
			sprintf(log, "[CC] Acceptor (%d) has reached the maximum number of connections [%d/%d]"
				, mAcceptor->getPort(), mAcceptor->getConnectNum(), mAcceptor->mMaxConnectNum);
			LoggerManager::getSingleton().log(log);
			closeSocket(recver);
			continue;
		}

		char clienthost[NI_MAXHOST] = {0x0};
		char clientservice[NI_MAXSERV] = {0x0};
		getnameinfo((sockaddr *)&clientaddr, addrlen, clienthost, sizeof(clienthost), clientservice, sizeof(clientservice), NI_NUMERICHOST|NI_NUMERICSERV);

	#if CELL_DEBUG_MODE
		char msg[128] = {0x0};
		sprintf(msg, "[CC] New connection: %s:%s @ %d", clienthost, clientservice, mAcceptor->getPort());
		LoggerManager::getSingleton().log(msg);
	#endif // CELL_DEBUG_MODE

		// 设置 socket 超时时间
		int timeout = BA_DEFAULT_TIMEOUT;
		if (::setsockopt(recver, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout)) == SOCKET_ERROR)
		{
			char log[128] = {0x0};
			sprintf(log, "[CC] Sets socket sending timeout failed - %s:%s", clienthost, clientservice);
			LoggerManager::getSingleton().log(log);
			closeSocket(recver);
			continue;
		}
		timeout = BA_DEFAULT_TIMEOUT;
		if (::setsockopt(recver, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout)) == SOCKET_ERROR)
		{
			char log[128] = {0x0};
			sprintf(log, "[CC] Sets socket receiving timeout failed - %s:%s", clienthost, clientservice);
			LoggerManager::getSingleton().log(log);
			closeSocket(recver);
			continue;
		}

		// 禁止发送合并的 Nagle 算法
		int optval = 1;
		if (::setsockopt(recver, IPPROTO_TCP, TCP_NODELAY, (char *)&optval, sizeof(optval)) == SOCKET_ERROR)
		{
			char log[128] = {0x0};
			sprintf(log, "[CC] Sets socket option TCP_NODELAY failed - %s:%s", clienthost, clientservice);
			LoggerManager::getSingleton().log(log);
			closeSocket(recver);
			continue;
		}

		// Keep alive
		int keepAlive = 1;
		if (::setsockopt(recver, SOL_SOCKET, SO_KEEPALIVE, (char *)&keepAlive, sizeof(keepAlive)) == SOCKET_ERROR)
		{
			char log[128] = {0x0};
			sprintf(log, "[CC] Sets socket option SO_KEEPALIVE failed - %s:%s", clienthost, clientservice);
			LoggerManager::getSingleton().log(log);
			closeSocket(recver);
			continue;
		}

		Session* session = mAcceptor->fireSessionCreated(recver, clienthost, atoi(clientservice));

		if (NULL != session)
		{
			if (false == mAcceptor->startRecvThread(session))
			{
				char log[128] = {0x0};
				sprintf(log, "[CC] Starts recv thread failed [%s:%s] @ %d", clienthost, clientservice, mAcceptor->getPort());
				LoggerManager::getSingleton().log(log);
				// Close socket
				closeSocket(recver);
			}
			else
			{
			#if CELL_DEBUG_MODE
				char msg[128] = {0x0};
				sprintf(msg, "[CC] Acceptor (%d) connection number : %d", mAcceptor->getPort(), mAcceptor->getConnectNum());
				LoggerManager::getSingleton().log(msg);
			#endif // CELL_DEBUG_MODE
			}
		}
		else
		{
			char log[128] = {0x0};
			sprintf(log, "[CC] Starts recv thread failed [%s:%s] @ %d", clienthost, clientservice, mAcceptor->getPort());
			LoggerManager::getSingleton().log(log);
			// Close socket
			closeSocket(recver);
		}
	}

	mRunning = false;
}
//-----------------------------------------------------------------------
void BADaemonThread::stop(void)
{
	mSpinning = false;
}


//-----------------------------------------------------------------------
BARecvThread::BARecvThread(BlockingAcceptor* acceptor, Session* session)
	: mAcceptor(acceptor)
	, mSession(session)
	, mBuffer(NULL)
	, mCacheCursor(0)
	, mRunning(false)
	, mSpinning(false)
	, expiredTime(0)
{
	mBuffer = CELL_ALLOC_T(char, BA_BUFSIZE, MEMCATEGORY_GENERAL);
	mCache = CELL_ALLOC_T(char, BA_CACHESIZE, MEMCATEGORY_GENERAL);
}
//-----------------------------------------------------------------------
BARecvThread::~BARecvThread(void)
{
	CELL_FREE(mCache, MEMCATEGORY_GENERAL);
	CELL_FREE(mBuffer, MEMCATEGORY_GENERAL);
}
//-----------------------------------------------------------------------
void BARecvThread::run(void)
{
	mRunning = true;

	mAcceptor->fireSessionOpened(mSession);

	socket_t recver = mAcceptor->getSock(mSession);

	const int size = mAcceptor->mBufSize;

	//char* data = NULL;

	mSpinning = true;
	while (mSpinning)
	{
		/*if (data != NULL)
		{
			CELL_FREE(data, MEMCATEGORY_GENERAL);
			data = NULL;
		}
		data = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(data, 0x0, size + 1);
		*/

		memset(mBuffer, 0x0, BA_BUFSIZE);
		char* data = mBuffer;

		int rsize = ::recv(recver, data, size, 0);

		if (SOCKET_ERROR == rsize)
		{
			int err = ::WSAGetLastError();
			switch (err)
			{
			case WSAETIMEDOUT:
				mSpinning = false;
				break;
			case WSAECONNRESET:
				mSpinning = false;
				break;
			case WSAECONNABORTED:
				mSpinning = false;
				break;
			default:
				// 连接错误
				mSpinning = false;
				break;
			}

		#if CELL_DEBUG_MODE
			if (err != WSAETIMEDOUT)
			{
				char log[128] = {0x0};
				sprintf(log, "[CC] Blocking acceptor (%d) thread close, error code: %d - [%s:%d]"
					, mAcceptor->getPort()
					, err
					, mSession->getAddress().getHostAddress().c_str()
					, mSession->getAddress().getPort());
				LoggerManager::getSingleton().log(log);
			}
		#endif

			continue;
		}
		else if (0 == rsize)
		{
			// 连接被关闭
			mSpinning = false;
			break;
		}

		if (mSpinning)
		{
			// 处理数据
			processRecv(data, rsize);
			//process(data, rsize);
		}
	}

	/*if (NULL != data)
	{
		CELL_FREE(data, MEMCATEGORY_GENERAL);
		data = NULL;
	}*/

#if CELL_DEBUG_MODE
	char msg[128] = {0x0};
	sprintf(msg, "[CC] Close connection: %s:%d @ %d"
		, mSession->getAddress().getHostAddress().c_str()
		, mSession->getAddress().getPort()
		, mAcceptor->getPort());
	LoggerManager::getSingleton().log(msg);
#endif

	//_ASSERTE(_CrtCheckMemory());

	// 关闭 Session
	mAcceptor->fireSessionClosed(mSession);

	mRunning = false;
}
//-----------------------------------------------------------------------
void BARecvThread::stop(void)
{
	mSpinning = false;
}
//-----------------------------------------------------------------------
void BARecvThread::processRecv(char* data, int size)
{
	if (mAcceptor->existDataMark())
	{
		if (mCacheCursor > 0 || (size_t)size <= mAcceptor->getHeadLength() + mAcceptor->getTailLength())
		{
			// 缓存

			int extractedLen = 0;

			if (mCacheCursor + size >= BA_CACHESIZE)
			{
				extractedLen = extractData(mCache, mCacheCursor);
			}
			else
			{
				char* cache = mCache;
				cache += mCacheCursor;
				memcpy(cache, data, size);
				mCacheCursor += size;

				extractedLen = extractData(mCache, mCacheCursor);
			}

			if (extractedLen > 0)
			{
				// FIXME 2012/5/4 需要继续判断是否完全提取完成
				//char buf[BA_CACHESIZE] = {0x0};
				//memcpy(buf, mCache + extractedLen, mCacheCursor - extractedLen);
				//memcpy(mCache, buf, mCacheCursor - extractedLen);

				memset(mCache, 0x0, BA_CACHESIZE);
				mCacheCursor = 0;
			}
			else
			{
				memset(mCache, 0x0, BA_CACHESIZE);
				mCacheCursor = 0;
			}
		}
		else
		{
			int extractedLen = extractData(data, size);
			if (extractedLen < size)
			{
				memcpy(mCache + mCacheCursor, data + extractedLen, size - extractedLen);
				mCacheCursor += size - extractedLen;
			}
		}
	}
	else
	{
		Message msg(data, size);
		mAcceptor->fireMessageReceived(mSession, &msg);
	}
}
//-----------------------------------------------------------------------
int BARecvThread::extractData(char* data, const int size)
{
	int extractedCursor = 0;

	char* dataptr = data;

	bool incomplete = false;

	while (!incomplete)
	{
		if (extractedCursor > size)
			break;

		bool valid = true;

		for (int i = 0, len = (int)mAcceptor->getHeadLength(); i < len; ++i)
		{
			if (mAcceptor->getHeadMark()[i] != dataptr[extractedCursor + i])
			{
				valid = false;
				break;
			}
		}

		if (valid)
		{
			extractedCursor += mAcceptor->getTailLength();

			int buflen = 0;

			// 开始解析数据
			while (true)
			{
				if (extractedCursor + buflen > size)
				{
					extractedCursor -= mAcceptor->getTailLength();
					incomplete = true;
					break;
				}

				// 判断结尾
				bool end = true;
				for (int i = 0, len = (int)mAcceptor->getTailLength(); i < len; ++i)
				{
					if (mAcceptor->getTailMark()[i] != dataptr[extractedCursor + buflen + i])
					{
						end = false;
						break;
					}
				}

				if (false == end)
				{
					// 数据没有结尾
					++buflen;
				}
				else
				{
					char buf[BA_CACHESIZE] = {0x0};
					memcpy(buf, dataptr + extractedCursor, buflen);

					Message msg(buf, buflen);
					mAcceptor->fireMessageReceived(mSession, &msg);

					extractedCursor += buflen;
					extractedCursor += mAcceptor->getTailLength();
					break;
				}
			}
		}
		else
		{
			break;
		}
	}

	return extractedCursor;
}
//-----------------------------------------------------------------------
void BARecvThread::process(char* data, int size)
{
	if (mAcceptor->existDataMark())
	{
		int extractedCursor = 0;
		if (mCacheCursor == 0)
		{
			// 没有被缓存的数据

			if ((size_t)size >= mAcceptor->getHeadLength())
			{
				// 数据长度大于起始标记长度
				// 提取数据
				extractedCursor = extract(data, size);
				if (extractedCursor == size)
				{
					// 数据完全提取完成
					// Nothing
				}
				else
				{
					// 没有提取全部数据
					// 从 extractedCursor 位置后的数据需要缓存，
					// 以便拼接上后续数据继续进行提取
					char* buf = data;
					buf += extractedCursor;
					cache(buf, size - extractedCursor);
				}
			}
			else
			{
				// 数据长度小于起始标记长度，
				// 放入缓存
				cache(data, size);
				// 尝试从缓存中提取数据
				extractedCursor = extract(mCache, mCacheCursor);
				if (extractedCursor == mCacheCursor)
				{
					// 缓存内的数据被全部提取
					clear();
				}
				else
				{
					// 缓存内的数据未被全部提取
					// 清除掉已经被提取的数据
					clear(extractedCursor);
				}
			}
		}
		else
		{
			// 当前系统有缓存数据
			// 将新数据写入缓存
			cache(data, size);
			// 尝试从缓存中提取数据
			extractedCursor = extract(mCache, mCacheCursor);
			if (extractedCursor == mCacheCursor)
			{
				// 缓存内的数据被全部提取
				clear();
			}
			else
			{
				// 缓存内的数据未被全部提取
				// 清除掉已经被提取的数据
				clear(extractedCursor);
			}
		}
	}
	else
	{
		Message msg(data, size);
		mAcceptor->fireMessageReceived(mSession, &msg);
	}
}
//-----------------------------------------------------------------------
int BARecvThread::extract(char* data, int size)
{
	char dataBuf[BA_CACHESIZE] = {0x0};
	int dataBufLength = 0;

	int retCursor = 0;
	char* dataptr = data;

	// 数据起始游标
	int dataBeginCursor = 0;
	// 是否是从检测阶段且回到提取阶段
	bool newData = false;

	int extractedCursor = 0;
	char phase = PARSE_PHASE_DISTINGUISH;

	while (extractedCursor < size)
	{
		switch (phase)
		{
		case PARSE_PHASE_DISTINGUISH:
		{
			// 查找起始标识

			// FIXME 2012/3/27
			//dataptr = data;
			//dataptr += extractedCursor;

			bool beginningValid = true;

			// 连续判断 Head Length 长度是否是数据起始标识域
			for (size_t i = 0, len = mAcceptor->getHeadLength(); i < len; ++i)
			{
				if (mAcceptor->getHeadMark()[i] != dataptr[i])
				{
					beginningValid = false;
					break;
				}
			}
			// 是合法的数据段头
			if (beginningValid)
			{
				extractedCursor += mAcceptor->getHeadLength();
				phase = PARSE_PHASE_EXTRACT;
				newData = true;

				// FIXME 2012/3/27
				dataptr += mAcceptor->getHeadLength();
			}
			else
			{
				++extractedCursor;

				// FIXME 2012/3/27
				dataptr += 1;
			}
			break;
		}
		case PARSE_PHASE_EXTRACT:
		{
			// FIXME 2012/3/27
			//dataptr = data;
			//dataptr += extractedCursor;

			// 记录数据起点
			if (newData)
			{
				newData = false;
				dataBeginCursor = extractedCursor;
				dataBufLength = 0;
			}

			bool endingValid = true;

			// 连续取 Tail Length 长度判断是否是数据结束标识域
			for (size_t i = 0, len = mAcceptor->getTailLength(); i < len; ++i)
			{
				if (mAcceptor->getTailMark()[i] != dataptr[i])
				{
					endingValid = false;
					break;
				}
			}

			if (false == endingValid)
			{
				// 记录原始数据长度
				++extractedCursor;
				++dataBufLength;

				// FIXME 2012/3/27
				dataptr += 1;
			}
			else
			{
				// 该段数据结束
				char* src = data;
				src += dataBeginCursor;
				memset(dataBuf, 0x0, BA_CACHESIZE);
				memcpy(dataBuf, src, dataBufLength);

				Message msg(dataBuf, dataBufLength);
				mAcceptor->fireMessageReceived(mSession, &msg);

				// 清空临时缓冲区
				memset(dataBuf, 0x0, BA_CACHESIZE);
				dataBufLength = 0;
				extractedCursor += mAcceptor->getTailLength();
				// 阶段置为识别
				phase = PARSE_PHASE_DISTINGUISH;
				// 更新返回游标
				retCursor = extractedCursor;

				// FIXME 2012/3/27
				dataptr += mAcceptor->getTailLength();
			}
			break;
		}
		default:
			break;
		}
	}

	return retCursor;
}
//-----------------------------------------------------------------------
void BARecvThread::cache(char* data, int size)
{
	size_t dataSize = mCacheCursor + size;
	// 如果数据长度大于缓冲区时，被当作无效数据丢弃
	if (dataSize > BA_CACHESIZE)
	{
		clear();
	}
	else
	{
		char* tail = mCache;
		tail += mCacheCursor;
		memcpy(tail, data, size);
		mCacheCursor += size;
	}
}
//-----------------------------------------------------------------------
void BARecvThread::clear(int cursor)
{
	if (cursor < 0)
	{
		memset(mCache, 0x0, BA_CACHESIZE);
		mCacheCursor = 0;
	}
	else
	{
		int size = mCacheCursor - cursor;

		char buf[BA_CACHESIZE] = {0x0};
		char* cacheBuf = mCache;
		cacheBuf += cursor;
		memcpy(buf, cacheBuf, size);

		memset(mCache, 0x0, BA_CACHESIZE);
		memcpy(mCache, buf, size);
		mCacheCursor = size;
	}
}

} // end namespace cell
